use std::any::{Any, TypeId};
use std::collections::{HashMap, HashSet};
use std::sync::{Arc, RwLock};
use std::time::Duration;

use async_trait::async_trait;
use tokio::sync::oneshot;
use tokio::time::timeout;
use uuid::Uuid;
use tracing::{debug, error, info, trace, warn};
use anyhow::{Result, anyhow};
use dashmap::DashMap;

use crate::actor::{Actor, ActorContext, ActorError, PID};
use crate::system::ActorSystem;

/// PubSub 系统错误类型
#[derive(Debug, thiserror::Error)]
pub enum PubSubError {
    #[error("Topic not found: {0}")]
    TopicNotFound(String),
    
    #[error("Subscribe failed: {0}")]
    SubscribeFailed(String),
    
    #[error("Publish failed: {0}")]
    PublishFailed(String),
    
    #[error("Request timeout")]
    Timeout,
    
    #[error("Message delivery failed: {0}")]
    MessageDeliveryFailed(String),
}

/// 发布/订阅系统
#[derive(Clone)]
pub struct PubSub {
    actor_system: Arc<ActorSystem>,
    topics: Arc<DashMap<String, Arc<Topic>>>,
    subscriber_timeout: Duration,
}

impl PubSub {
    /// 创建新的发布/订阅系统
    pub fn new(actor_system: Arc<ActorSystem>) -> Self {
        Self {
            actor_system,
            topics: Arc::new(DashMap::new()),
            subscriber_timeout: Duration::from_secs(5),
        }
    }
    
    /// 设置订阅者超时时间
    pub fn with_subscriber_timeout(mut self, timeout: Duration) -> Self {
        self.subscriber_timeout = timeout;
        self
    }
    
    /// 启动发布/订阅系统
    pub async fn start(&self) -> Result<()> {
        // 创建PubSubDeliveryActor
        let delivery_props = crate::actor::Props::new(Box::new(|| {
            Box::new(PubSubDeliveryActor::new(self.subscriber_timeout))
        }));
        
        let _ = self.actor_system.spawn_named(delivery_props, "pubsub-delivery").await?;
        info!("Started PubSub system");
        Ok(())
    }
    
    /// 订阅主题
    pub async fn subscribe(&self, topic_name: &str, subscriber: PID) -> Result<()> {
        let topic = self.get_or_create_topic(topic_name);
        topic.add_subscriber(subscriber).await?;
        Ok(())
    }
    
    /// 取消订阅
    pub async fn unsubscribe(&self, topic_name: &str, subscriber: &PID) -> Result<()> {
        if let Some(topic) = self.topics.get(topic_name) {
            topic.remove_subscriber(subscriber).await?;
            Ok(())
        } else {
            Err(anyhow!(PubSubError::TopicNotFound(topic_name.to_string())))
        }
    }
    
    /// 发布消息到主题
    pub async fn publish<T: Any + Send + 'static>(&self, topic_name: &str, message: T) -> Result<()> {
        let topic = self.get_or_create_topic(topic_name);
        topic.publish(Box::new(message)).await?;
        Ok(())
    }
    
    /// 发布带有请求/响应的消息
    pub async fn publish_request<T: Any + Send + 'static, R: Any + Send + 'static>(
        &self, 
        topic_name: &str, 
        message: T,
        timeout_duration: Duration
    ) -> Result<R> {
        let topic = self.get_or_create_topic(topic_name);
        let response = topic.publish_request::<R>(Box::new(message), timeout_duration).await?;
        Ok(response)
    }
    
    /// 获取或创建主题
    fn get_or_create_topic(&self, topic_name: &str) -> Arc<Topic> {
        if let Some(topic) = self.topics.get(topic_name) {
            topic.clone()
        } else {
            let topic = Arc::new(Topic::new(
                topic_name.to_string(),
                self.actor_system.clone()
            ));
            self.topics.insert(topic_name.to_string(), topic.clone());
            topic
        }
    }
}

/// 主题
pub struct Topic {
    name: String,
    subscribers: RwLock<HashSet<PID>>,
    actor_system: Arc<ActorSystem>,
}

impl Topic {
    /// 创建新主题
    fn new(name: String, actor_system: Arc<ActorSystem>) -> Self {
        Self {
            name,
            subscribers: RwLock::new(HashSet::new()),
            actor_system,
        }
    }
    
    /// 添加订阅者
    async fn add_subscriber(&self, subscriber: PID) -> Result<()> {
        let mut subscribers = self.subscribers.write().unwrap();
        subscribers.insert(subscriber);
        debug!("Added subscriber to topic {}", self.name);
        Ok(())
    }
    
    /// 移除订阅者
    async fn remove_subscriber(&self, subscriber: &PID) -> Result<()> {
        let mut subscribers = self.subscribers.write().unwrap();
        subscribers.remove(subscriber);
        debug!("Removed subscriber from topic {}", self.name);
        Ok(())
    }
    
    /// 发布消息
    async fn publish(&self, message: Box<dyn Any + Send>) -> Result<()> {
        let subscribers = self.subscribers.read().unwrap();
        if subscribers.is_empty() {
            debug!("No subscribers for topic {}", self.name);
            return Ok(());
        }
        
        // 创建消息批次
        let batch = PubSubBatch {
            topic: self.name.clone(),
            message,
            trace_id: Uuid::new_v4().to_string(),
        };
        
        // 使用PubSubDeliveryActor发送批次
        let delivery_pid = PID::new("".to_string(), "pubsub-delivery".to_string());
        
        // 创建DeliverBatchRequest
        let deliver_request = DeliverBatchRequest {
            batch,
            subscribers: subscribers.clone(),
        };
        
        // 发送消息
        self.actor_system.send(&delivery_pid, deliver_request).await?;
        Ok(())
    }
    
    /// 发布带有请求/响应的消息
    async fn publish_request<R: Any + Send + 'static>(
        &self, 
        message: Box<dyn Any + Send>,
        timeout_duration: Duration
    ) -> Result<R> {
        let subscribers = self.subscribers.read().unwrap();
        if subscribers.is_empty() {
            return Err(anyhow!(PubSubError::PublishFailed("No subscribers".to_string())));
        }
        
        // 创建oneshot通道用于接收响应
        let (tx, rx) = oneshot::channel();
        
        // 创建RequestMessage，封装原始消息和发送者
        let request_message = RequestEnvelope {
            message,
            sender: tx,
            trace_id: Uuid::new_v4().to_string(),
            type_tag: std::any::type_name::<R>().to_string(),
        };
        
        // 创建消息批次
        let batch = PubSubBatch {
            topic: self.name.clone(),
            message: Box::new(request_message),
            trace_id: Uuid::new_v4().to_string(),
        };
        
        // 使用PubSubDeliveryActor发送批次
        let delivery_pid = PID::new("".to_string(), "pubsub-delivery".to_string());
        
        // 创建DeliverBatchRequest
        let deliver_request = DeliverBatchRequest {
            batch,
            subscribers: subscribers.clone(),
        };
        
        // 发送消息
        self.actor_system.send(&delivery_pid, deliver_request).await?;
        
        // 等待响应或超时
        match timeout(timeout_duration, rx).await {
            Ok(result) => {
                match result {
                    Ok(response) => {
                        // 尝试将响应转换为期望的类型
                        match response.downcast::<R>() {
                            Ok(typed_response) => Ok(*typed_response),
                            Err(_) => Err(anyhow!(PubSubError::PublishFailed("Response type mismatch".to_string())))
                        }
                    },
                    Err(_) => Err(anyhow!(PubSubError::PublishFailed("Sender dropped".to_string())))
                }
            },
            Err(_) => Err(anyhow!(PubSubError::Timeout))
        }
    }
}

/// 消息批次
#[derive(Clone)]
pub struct PubSubBatch {
    pub topic: String,
    pub message: Box<dyn Any + Send>,
    pub trace_id: String,
}

/// 发送批次请求
#[derive(Clone)]
pub struct DeliverBatchRequest {
    pub batch: PubSubBatch,
    pub subscribers: HashSet<PID>,
}

/// 请求消息包装
pub struct RequestEnvelope {
    pub message: Box<dyn Any + Send>,
    pub sender: oneshot::Sender<Box<dyn Any + Send>>,
    pub trace_id: String,
    pub type_tag: String,
}

/// 发布/订阅消息传递Actor
pub struct PubSubDeliveryActor {
    subscriber_timeout: Duration,
}

impl PubSubDeliveryActor {
    /// 创建新的发布/订阅消息传递Actor
    pub fn new(subscriber_timeout: Duration) -> Self {
        Self {
            subscriber_timeout,
        }
    }
    
    /// 传递消息批次
    async fn deliver_batch(&self, ctx: &mut impl ActorContext, batch: &PubSubBatch, subscriber: &PID) -> Result<()> {
        ctx.send(subscriber, batch.clone()).await?;
        Ok(())
    }
    
    /// 传递请求消息批次并等待响应
    async fn deliver_request_batch(&self, ctx: &mut impl ActorContext, batch: &PubSubBatch, subscriber: &PID) -> Result<Box<dyn Any + Send>> {
        let response = ctx.request::<Box<dyn Any + Send>>(subscriber, batch.clone(), self.subscriber_timeout).await?;
        Ok(response)
    }
}

#[async_trait]
impl Actor for PubSubDeliveryActor {
    type Context = crate::actor::DefaultActorContext;
    
    async fn receive(&self, ctx: &mut Self::Context, msg: Box<dyn Any + Send>) -> Result<(), ActorError> {
        if let Some(deliver_req) = msg.downcast_ref::<DeliverBatchRequest>() {
            let batch = &deliver_req.batch;
            let subscribers = &deliver_req.subscribers;
            
            // 检查是否是请求消息
            let is_request = batch.message.type_id() == TypeId::of::<RequestEnvelope>();
            
            if is_request {
                if let Some(envelope) = batch.message.downcast_ref::<RequestEnvelope>() {
                    // 对于请求消息，我们只向第一个订阅者发送消息
                    if let Some(first_subscriber) = subscribers.iter().next() {
                        match self.deliver_request_batch(ctx, batch, first_subscriber).await {
                            Ok(response) => {
                                // 尝试发送响应
                                if let Some(envelope) = batch.message.downcast_ref::<RequestEnvelope>() {
                                    let _ = envelope.sender.send(response);
                                }
                            },
                            Err(e) => {
                                error!("Failed to deliver request batch: {}", e);
                            }
                        }
                    }
                }
            } else {
                // 对于普通消息，我们向所有订阅者发送消息
                for subscriber in subscribers {
                    if let Err(e) = self.deliver_batch(ctx, batch, subscriber).await {
                        error!("Failed to deliver batch to {}: {}", subscriber.id, e);
                    }
                }
            }
        }
        
        Ok(())
    }
}

/// 发布者
pub struct Publisher {
    pub_sub: PubSub,
    topic: String,
}

impl Publisher {
    /// 创建新的发布者
    pub fn new(pub_sub: PubSub, topic: String) -> Self {
        Self {
            pub_sub,
            topic,
        }
    }
    
    /// 发布消息
    pub async fn publish<T: Any + Send + 'static>(&self, message: T) -> Result<()> {
        self.pub_sub.publish(&self.topic, message).await
    }
    
    /// 发布带有请求/响应的消息
    pub async fn publish_request<T: Any + Send + 'static, R: Any + Send + 'static>(
        &self, 
        message: T,
        timeout: Duration
    ) -> Result<R> {
        self.pub_sub.publish_request::<T, R>(&self.topic, message, timeout).await
    }
}

/// 订阅者
pub struct Subscriber {
    pub_sub: PubSub,
    topic: String,
    subscriber_pid: PID,
}

impl Subscriber {
    /// 创建新的订阅者
    pub fn new(pub_sub: PubSub, topic: String, subscriber_pid: PID) -> Self {
        Self {
            pub_sub,
            topic,
            subscriber_pid,
        }
    }
    
    /// 订阅主题
    pub async fn subscribe(&self) -> Result<()> {
        self.pub_sub.subscribe(&self.topic, self.subscriber_pid.clone()).await
    }
    
    /// 取消订阅
    pub async fn unsubscribe(&self) -> Result<()> {
        self.pub_sub.unsubscribe(&self.topic, &self.subscriber_pid).await
    }
} 